/**
 * @file event_bus.cpp
 * @brief EventBus 类核心实现。
 */

#include "event_bus.h"

#include <algorithm>

EventBus::EventBus()
    : worker_([this] { DispatchLoop(); })  // 启动分发线程。
{}

EventBus::~EventBus() {
    Stop();
}

Subscription EventBus::Subscribe(std::string topic, Callback cb) {
    const auto id = next_id_.fetch_add(1, std::memory_order_relaxed);

    {
        std::unique_lock lock(subs_mutex_);
        subscribers_[topic].push_back(Subscriber{id, std::move(cb)});
    }

    return Subscription{this, std::move(topic), id};
}

void EventBus::Publish(std::string topic, std::string payload) {
    queue_.Push(Message{std::move(topic), std::move(payload)});
}

void EventBus::Stop() {
    bool expected = false;
    if (stopped_.compare_exchange_strong(expected, true)) {
        // 让阻塞队列退出，并等待工作线程结束。
        queue_.Stop();
        if (worker_.joinable()) {
            worker_.join();
        }

        // 清空订阅者，释放回调。
        std::unique_lock lock(subs_mutex_);
        subscribers_.clear();
    }
}

void EventBus::Unsubscribe(const std::string& topic, std::size_t id) {
    std::unique_lock lock(subs_mutex_);
    if (auto it = subscribers_.find(topic); it != subscribers_.end()) {
        auto& vec = it->second;
        vec.erase(
            std::remove_if(vec.begin(), vec.end(),
                           [&](const Subscriber& sub) { return sub.id == id; }),
            vec.end());

        if (vec.empty()) {
            subscribers_.erase(it);
        }
    }
}

void EventBus::DispatchLoop() {
    while (auto message = queue_.Pop()) {
        std::vector<Callback> callbacks;

        {
            // 读取订阅者列表，可与其他读取并行。
            std::shared_lock lock(subs_mutex_);
            if (auto it = subscribers_.find(message->topic);
                it != subscribers_.end()) {
                callbacks.reserve(it->second.size());
                for (auto& sub : it->second) {
                    callbacks.push_back(sub.cb);
                }
            }
        }

        // 在锁外执行回调，避免用户回调阻塞导致锁被占用。
        for (auto& cb : callbacks) {
            cb(*message);
        }
    }
}